package com.ds.lens.flink.monitor.monitorjob.function;

import com.ds.lens.data.common.client.LensMonitorClient;
import com.ds.lens.data.common.client.vo.LensApplicationNameOutVO;
import com.ds.lens.data.common.client.vo.MonitorKeyOutVO;
import com.ds.lens.data.common.document.dto.MonitorKey;
import com.ds.lens.flink.monitor.MonitorJobContext;
import com.ds.lens.flink.monitor.common.bo.MonitorState;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.springframework.util.CollectionUtils;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

@Slf4j
public class LensMonitorSource implements SourceFunction<MonitorState> {
    private volatile boolean isRunning = true;
    private final long period;
    private final TimeUnit unit;
    private Long lastModifyMonitorTime;
    private Long lastModifyApplNameTime;
    private ConcurrentHashMap<Long, MonitorState> monitorStateMap = new ConcurrentHashMap<>(5000);
    private ConcurrentHashMap<Long, LensApplicationNameOutVO> applicationNameMap = new ConcurrentHashMap<>(1000);

    public LensMonitorSource(long period, TimeUnit unit) {
        this.period = period;
        this.unit = unit;
    }

    @Override
    public void run(SourceFunction.SourceContext<MonitorState> ctx) throws Exception {
        while (isRunning) {
            try {
                updateApplicationName();
                List<MonitorState> deletedStates = updateMonitorStateMap();
                for (Long key : monitorStateMap.keySet()) {
                    MonitorState state = monitorStateMap.get(key);
                    state.setApplicationName(applicationNameMap.get(state.getApplicationId()).getName());
                    ctx.collect(monitorStateMap.get(key));
                }
                for (MonitorState state : deletedStates) {
                    ctx.collect(state);
                }
                Thread.sleep(unit.toMillis(period));
            } catch (Exception e) {
                log.error("error occurred. ", e);
            }
        }
    }

    @Override
    public void cancel() {
        isRunning = false;
    }

    private List<MonitorState> updateMonitorStateMap() {
        LensMonitorClient monitorClient = MonitorJobContext.getInstance().getMonitorClient();
        List<MonitorKeyOutVO> monitorIds;
        if (lastModifyMonitorTime == null) {
            monitorIds = monitorClient.listAvailableMonitorState().getData();
        } else {
            monitorIds = monitorClient.listUpdatedMonitorState(lastModifyMonitorTime).getData();
        }
        List<MonitorState> deletedStates = new ArrayList<>();
        if (!CollectionUtils.isEmpty(monitorIds)) {
            Long max = null;
            for (MonitorKeyOutVO outVO : monitorIds) {
                MonitorState state = MonitorState.builder()
                        .key(new MonitorKey(outVO.getId(), outVO.getVersion()))
                        .applicationId(outVO.getApplicationId())
                        .modifyTime(outVO.getModifiedTime())
                        .isDelete(outVO.getIsDelete())
                        .runStatus(outVO.getRunStatus())
                        .build();
                if (max == null || max < outVO.getModifiedTime()) {
                    max = outVO.getModifiedTime();
                }
                if (!state.getIsDelete()) {
                    monitorStateMap.put(state.getKey().getMonitorId(), state);
                } else {
                    monitorStateMap.remove(state.getKey().getMonitorId());
                    deletedStates.add(state);
                }
            }
            lastModifyMonitorTime = max;
        }
        return deletedStates;
    }

    private void updateApplicationName() {
        LensMonitorClient monitorClient = MonitorJobContext.getInstance().getMonitorClient();
        List<LensApplicationNameOutVO> appNames;
        if (lastModifyApplNameTime == null) {
            appNames = monitorClient.listAllApplications().getData();
        } else {
            appNames = monitorClient.listUpdatedApplications(lastModifyApplNameTime).getData();
        }
        if (!CollectionUtils.isEmpty(appNames)) {
            Long max = null;
            for (LensApplicationNameOutVO outVO : appNames) {
                if (max == null || max < outVO.getModifiedTime()) {
                    max = outVO.getModifiedTime();
                }
                applicationNameMap.put(outVO.getId(), outVO);
            }
            lastModifyApplNameTime = max;
        }
    }
}
